package edu.tyut.game;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.randInteger;

// 床前明月光
public class GameStreamUpsert {

    final static Logger logger = LoggerFactory.getLogger(GameStreamUpsert.class);

    @SuppressWarnings("Duplicates")
    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 2. 读取数据源 TODO
        DataStreamSource<String> lineDS = env.socketTextStream("localhost", 7778);
        lineDS.print();
        logger.info(lineDS.toString());


        SingleOutputStreamOperator<Tuple5<String, Integer, Integer, Integer, Long>> operatorFlyGame = lineDS.filter(it -> it.contains("flyGame"))
                .map(it -> {
                    String[] str = it.split(",");
                    return new Tuple5<>(str[0], Integer.parseInt(str[1]), Integer.parseInt(str[2]), Integer.parseInt(str[3]), Long.parseLong(str[4]));
                }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT, Types.LONG));

        SingleOutputStreamOperator<Tuple6<String, Integer, Integer, Integer, Integer, Long>> operatorPoemRecommend = lineDS.filter(it -> it.contains("PoemRecommend"))
                .map(it -> {
                    String[] str = it.split(",");
                    return new Tuple6<>(str[0], Integer.parseInt(str[1]), Integer.parseInt(str[2]), Integer.parseInt(str[3]), Integer.parseInt(str[4]), Long.parseLong(str[5]));
                }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT, Types.INT, Types.LONG));

        // 3. 创建Schema
        final Schema schemaFlyGame = Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .column("f1", DataTypes.INT())
                .column("f2", DataTypes.INT())
                .column("f3", DataTypes.INT())
                .column("f4", DataTypes.BIGINT())
                .build();

        final Schema schemaPoemRecommend = Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .column("f1", DataTypes.INT())
                .column("f2", DataTypes.INT())
                .column("f3", DataTypes.INT())
                .column("f4", DataTypes.INT())
                .column("f5", DataTypes.BIGINT())
                .build();

        // 4. 临时表
        String flyGameView = "flyGameView";
        String poemRecommendView = "poemRecommendView";

        tEnv.createTemporaryView(flyGameView, operatorFlyGame, schemaFlyGame);
        tEnv.createTemporaryView(poemRecommendView, operatorPoemRecommend, schemaPoemRecommend);

        // 5. 创建表
        Table flyGameTable = tEnv.sqlQuery("select f0 as flyGame, f1 as uid, f2 as total_rounds, f3 as win_rounds, f4 as time_stamp from flyGameView");
        // flyGameTable.execute().print();
        Table flyGameTableResult = flyGameTable
                .groupBy($("flyGame"), $("uid"))
                // , String.valueOf($("输赢(0,1)").dividedBy($("总局数")).as("胜率"))
                .select($("flyGame"), $("uid"), $("total_rounds").sum().as("total_rounds"), $("win_rounds").sum().as("win_rounds"))
                .select($("flyGame"), $("uid"), $("total_rounds"), $("win_rounds"),
                        $("win_rounds").cast(DataTypes.DOUBLE()).dividedBy($("total_rounds").cast(DataTypes.DOUBLE())).as("win_rate"),
                        $("total_rounds").minus($("win_rounds")).as("lose_rounds"))
                .select(randInteger(1).as("id"), $("uid"), $("total_rounds"), $("win_rounds"), $("win_rate"))
                .distinct();

        Table poemRecommendViewTable = tEnv.sqlQuery("select f0 as PoemRecommend, f1 as uid, f2 as poemId, f3 as poetId, f4 as click_count, f5 as time_stamp from poemRecommendView");
        // poemRecommendViewTable.execute().print();
        Table poemRecommendResult = poemRecommendViewTable
                .groupBy($("PoemRecommend"), $("uid"), $("poemId"))
                .select($("PoemRecommend"), $("uid"), $("poemId"), $("click_count").sum().as("click_count"))
                .select(randInteger(1).as("id"), $("poemId"), $("click_count"))
                .distinct();

        // 6. 使用Flink upsert 模式
        tEnv.executeSql("create table if not exists tb_fly_flower_order(" +
                "id int," +
                "uid int," +
                "total_rounds int," +
                "win_rounds int," +
                "win_rate double," +
                "primary key (uid) not enforced" +
                ")with(" +
                "'connector' = 'jdbc'," +
                "'driver' = 'com.mysql.cj.jdbc.Driver'," +
                "'url' = 'jdbc:mysql://localhost:3306/db_game?serverTimezone=Asia/Shanghai'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'connection.max-retry-timeout' = '60s'," +
                "'table-name' = 'tb_fly_flower_order'," +
                "'sink.buffer-flush.max-rows' = '500'," +
                "'sink.buffer-flush.interval' = '5s'," +
                "'sink.max-retries' = '3'," +
                "'sink.parallelism' = '1');");

        tEnv.executeSql("create table if not exists tb_poem_recommend(" +
                "id int," +
                "poem_id int," +
                "click_count bigint," +
                "primary key (poem_id) not enforced" +
                ")with(" +
                "'connector' = 'jdbc'," +
                "'driver' = 'com.mysql.cj.jdbc.Driver'," +
                "'url' = 'jdbc:mysql://localhost:3306/db_game?serverTimezone=Asia/Shanghai'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'connection.max-retry-timeout' = '60s'," +
                "'table-name' = 'tb_poem_recommend'," +
                "'sink.buffer-flush.max-rows' = '500'," +
                "'sink.buffer-flush.interval' = '5s'," +
                "'sink.max-retries' = '3'," +
                "'sink.parallelism' = '1');");


        // 7. 插入数据库

        TablePipeline tablePipeline1 = flyGameTableResult.insertInto("tb_fly_flower_order");
        TablePipeline tablePipeline2 = poemRecommendResult.insertInto("tb_poem_recommend");

        StatementSet statementSet = tEnv.createStatementSet();
        statementSet.add(tablePipeline1);
        statementSet.add(tablePipeline2);
        statementSet.execute().print();


        Table oederTable = tEnv.from("tb_fly_flower_order");
        Table poemTable = tEnv.from("tb_poem_recommend");

        tEnv.toChangelogStream(oederTable).print();
        tEnv.toChangelogStream(poemTable).print();

        // 8. 执行
        env.execute("Job");
    }
}
